package org.jeecg.modules.syncdata.handler;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSONObject;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.annotation.XxlJob;
import com.xxl.job.core.util.ShardingUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.jeecg.modules.syncdata.dto.SyncDataDto;
import org.jeecg.modules.syncdata.service.SyncCompensateService;
import org.jeecg.modules.syncdata.service.SyncLogService;
import org.jeecg.modules.syncdata.service.SyncService;
import org.jeecg.modules.thirdpartyapi.constant.BgApiEnum;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import static com.xxl.job.core.biz.model.ReturnT.FAIL;
import static com.xxl.job.core.biz.model.ReturnT.SUCCESS;

/**
 * 同步处理程序
 *
 * @author zhanglei01
 * @date 2022/03/16
 */
@Slf4j
@Component

public class SyncHandler {

    @Value("${xxl.job.executor.ip}")
    private String ip;

    @Autowired
    private SyncService syncService;

    @Autowired
    private SyncCompensateService syncCompensateService;

    @Autowired
    private SyncLogService syncLogService;

    @XxlJob(value = "syncHandler")
    public ReturnT<String> execute(String jobParam) throws Exception {
        //处理请求参数
        SyncDataDto syncDataDto = null;
        if (StringUtils.isNotBlank(jobParam)) {
            syncDataDto = JSONObject.parseObject(jobParam, SyncDataDto.class);
        }
        //如果入参有设置表名，只同步设置的表，没有设置就同步BgApiEnum中的所有表数据
        BgApiEnum[] values = null;
        if (StringUtils.isNotBlank(syncDataDto.getTableName())) {
            for (BgApiEnum value : BgApiEnum.values()) {
                if (value.getValue().equals(syncDataDto.getTableName())) {
                    values = new BgApiEnum[1];
                    values[0] = value;
                    break;
                }
            }
        }
        //没有设置就同步BgApiEnum中的所有表数据
        if (values == null || values.length == 0) {
            values = BgApiEnum.values();
            syncDataDto.setTableName("");
        }
        //没有设置时间，获取最后同步时间
        if (StringUtils.isBlank(syncDataDto.getStartTime()) || StringUtils.isBlank(syncDataDto.getEndTime())) {
            //获取任务日志表最后一次同步的日期的下一天N，如果是昨天之前，先同步N,否则同步昨天数据
            DateTime syncDateTime = syncLogService.getSyncDateTime(syncDataDto.getTableName());
            syncDataDto.setStartTime(DateUtil.beginOfDay(syncDateTime).toString());
            syncDataDto.setEndTime(DateUtil.endOfDay(syncDateTime).toString());
        }
        String dayTime = DateUtil.parseDate(syncDataDto.getStartTime()).toDateStr();
        if (dayTime.equals(DateUtil.today())) {
            log.error("同步时间错误！");
            return FAIL;
        }
        if (syncLogService.getSyncResult(dayTime)) {
            log.info("全部数据已同步完成！");
            return SUCCESS;
        }
        //获取分片信息
        ShardingUtil.ShardingVO shardingVo = ShardingUtil.getShardingVo();
        if (shardingVo == null) {
            shardingVo = new ShardingUtil.ShardingVO(0, 1);
        }
        log.info("SyncHandler同步数据,分片参数：当前分片序号 = {}, 总分片数 = {},ip={}", shardingVo.getIndex(), shardingVo.getTotal(), ip);

        ExecutorService executor = new ThreadPoolExecutor(5, 5, 0L,
                TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(),
                new ThreadPoolExecutor.CallerRunsPolicy());
        //判断当天除order外是否全部完成
        Boolean syncResultExceptOrder = syncLogService.getSyncResultExceptOrder(dayTime);
        if (!syncResultExceptOrder) {
            for (BgApiEnum bgApiEnum : values) {
                if (bgApiEnum.equals(BgApiEnum.sjmkjg_order) || bgApiEnum.equals(BgApiEnum.swzbjc_order)) {
                    continue;
                }
                //分片，只同步命中的
                if (bgApiEnum.getSort() % shardingVo.getTotal() != shardingVo.getIndex()) {
                    continue;
                }
                log.info("SyncHandler同步：{}表数据,第 {} 片, 命中分片开始处理", bgApiEnum.getValue(), shardingVo.getIndex());
                syncService.syncSingleTable(executor, bgApiEnum, syncDataDto, dayTime);
            }
        }
        //判断前一天和当天除order外是否全部同步完成，全部完成同步order表
        syncResultExceptOrder = syncLogService.getBeforeDaySyncResult(dayTime);
        if (syncResultExceptOrder) {
            if (syncLogService.getSyncResult(dayTime)) {
                executor.shutdown();
                log.info("全部数据已同步完成");
                return SUCCESS;
            }
            //同步order表
            ArrayList<BgApiEnum> bgApi = new ArrayList<>();
            bgApi.add(BgApiEnum.sjmkjg_order);
            bgApi.add(BgApiEnum.swzbjc_order);
            for (BgApiEnum bgApiEnum : bgApi) {
                //分片，只同步命中的
                if (bgApiEnum.getSort() % shardingVo.getTotal() != shardingVo.getIndex()) {
                    continue;
                }
                syncService.syncSingleTable(executor, bgApiEnum, syncDataDto, dayTime);
            }
        }
        executor.shutdown();
        log.info("同步数据执行完成");
        return SUCCESS;
    }

}
